package com.sdses.flink.apps;

import com.alibaba.fastjson2.JSONObject;
import com.ververica.cdc.connectors.base.options.StartupOptions;
import com.ververica.cdc.connectors.base.source.jdbc.JdbcIncrementalSource;
import com.ververica.cdc.connectors.postgres.source.PostgresSourceBuilder;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.PreparedStatement;
import java.sql.SQLException;


public class PgCdc2Pg2 {
    public static final Logger logger = LoggerFactory.getLogger(PgCdc2Pg2.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.enableCheckpointing(30000);

        //Source
        JdbcIncrementalSource<DebeziumRecord> postgresIncrementalSource =
                PostgresSourceBuilder.PostgresIncrementalSource.<DebeziumRecord>builder()
                        .hostname("192.168.102.154")
                        .port(5866)
                        .database("db_datacube_sdses_test")
                        .schemaList("public")
                        .tableList("public.test_gis_erp_community2")
                        .username("sysdba")
                        .password("Sdses,hg2o23")
                        .slotName("flink")
                        .decodingPluginName("pgoutput")
                        .deserializer(new CustomDebeziumDeserializationSchema())
                        .includeSchemaChanges(true)
                        .startupOptions(StartupOptions.initial())
                        .build();


        //Sink
        env.fromSource(postgresIncrementalSource, WatermarkStrategy.noWatermarks(), "PostgreSQLCDC")
                .addSink(JdbcSink.sink(
                        "WITH input_data (code, name, id, shape_length, shape_area, the_geom, import_time_, operation) AS ( " +
                                "VALUES (?,?,?,?,?,ST_GeomFromText(?),TO_TIMESTAMP(?, 'YYYY-MM-DD HH24:MI:SS.US'),?) " +
                                "), " +
                                "update_operation AS ( " +
                                "UPDATE public.test_gis_erp_community_sink " +
                                "SET name = input.name, code = input.code, shape_length = input.shape_length, shape_area = input.shape_area, the_geom = input.the_geom, import_time_ = input.import_time_ " +
                                "FROM input_data input " +
                                "WHERE public.test_gis_erp_community_sink.id = input.id AND input.operation = 'U' " +
                                "RETURNING public.test_gis_erp_community_sink.* " +
                                "), " +
                                "insert_operation AS ( " +
                                "INSERT INTO public.test_gis_erp_community_sink (code, name, id, shape_length, shape_area, the_geom, import_time_) " +
                                "SELECT code, name, id, shape_length, shape_area, the_geom, import_time_ " +
                                "FROM input_data " +
                                "WHERE operation = 'I' AND NOT EXISTS (SELECT 1 FROM update_operation WHERE update_operation.id = input_data.id) " +
                                "RETURNING public.test_gis_erp_community_sink.* " +
                                "), " +
                                "delete_operation AS ( " +
                                "DELETE FROM public.test_gis_erp_community_sink " +
                                "WHERE id IN (SELECT id FROM input_data WHERE operation = 'D') " +
                                "RETURNING public.test_gis_erp_community_sink.* " +
                                ") " +
                                "SELECT * FROM update_operation " +
                                "UNION ALL " +
                                "SELECT * FROM insert_operation " +
                                "UNION ALL " +
                                "SELECT * FROM delete_operation",
                        new JdbcStatementBuilder<DebeziumRecord>() {
                            @Override
                            public void accept(PreparedStatement pStmt, DebeziumRecord record) throws SQLException {
                                JSONObject after = record.getAfter();
                                String recordOp = record.getOp();
                                JSONObject before = record.getBefore();

                                String operation = "";
                                if (after!= null && recordOp.equalsIgnoreCase("U")) {
                                    operation = "U";  // 假设after不为null表示更新操作，这里可以根据实际更精准判断，比如从record中获取具体操作标识字段
                                    logger.info("PgCdc2Pg任务执行insert/update操作...");
                                    pStmt.setString(1, after.getString("code"));
                                    pStmt.setString(2, after.getString("name"));
                                    pStmt.setInt(3, after.getIntValue("id"));
                                    pStmt.setDouble(4, after.getDouble("shape_length"));
                                    pStmt.setDouble(5, after.getDouble("shape_area"));
                                    pStmt.setString(6, after.getString("the_geom"));
                                    pStmt.setString(7, after.getString("import_time_"));
                                    pStmt.setString(8, operation);
                                } else if (after!= null && recordOp.equalsIgnoreCase("I")){
                                    operation = "I";  // 可根据实际情况补充更准确的插入操作判断逻辑
                                    logger.info("PgCdc2Pg任务执行insert操作...");
                                    // 这里假设插入操作时相关字段从某个地方获取，需要根据实际调整，比如从record中获取对应值
                                    pStmt.setString(1, "default_code");
                                    pStmt.setString(2, "default_name");
                                    pStmt.setInt(3, 0);
                                    pStmt.setDouble(4, 0.0);
                                    pStmt.setDouble(5, 0.0);
                                    pStmt.setString(6, "default_geom");
                                    pStmt.setString(7, "default_import_time");
                                    pStmt.setString(8, operation);
                                } else if (before!= null && recordOp.equalsIgnoreCase("D")){
                                    operation = "D";  // 假设after为null且before不为null表示删除操作，同样可按需精准判断
                                    logger.info("PgCdc2Pg任务执行delete操作...");
                                    pStmt.setString(1, before.getString("code"));
                                    pStmt.setString(2, before.getString("name"));
                                    pStmt.setInt(3, before.getIntValue("id"));
                                    pStmt.setDouble(4, before.getDouble("shape_length"));
                                    pStmt.setDouble(5, before.getDouble("shape_area"));
                                    pStmt.setString(6, before.getString("the_geom"));
                                    pStmt.setString(7, before.getString("import_time_"));
                                    pStmt.setString(8, operation);
                                }
                            }
                        },
                        JdbcExecutionOptions.builder()
                                .withBatchSize(10) // 批次大小,条数
                                .withBatchIntervalMs(5000) // 批次最大等待时间
                                .withMaxRetries(1) // 重复次数
                                .build(), // 写入参数配置,
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUrl("jdbc:postgresql://192.168.102.154:5866/db_datacube_sdses_test")
                                .withUsername("sysdba")
                                .withPassword("Sdses,hg2o23")
                                .withDriverName("org.postgresql.Driver")
                                .build()));

        env.execute();
    }
}
